package com.hujf.flink.stock;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author hujf
 * @Date 2022/4/18 15:03
 * @Description TODO
 */
public class SoutceTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

       // environment.setStreamTimeCharacteristic();

      //  environment.getConfig().setAutoWatermarkInterval(1000);


        StockSource stockSource = new StockSource("stock");

        DataStreamSource<StockPrice> stockPriceDataStreamSource = environment.addSource(stockSource);

        SingleOutputStreamOperator<Tuple2<String,String>> objectSingleOutputStreamOperator = stockPriceDataStreamSource.flatMap(new FlatMapFunction<StockPrice, Tuple2<String,String>>() {
            @Override
            public void flatMap(StockPrice value, Collector<Tuple2<String,String>> out) throws Exception {
                out.collect(Tuple2.of(value.symbol, value.price));
            }
        });
        objectSingleOutputStreamOperator.print();
        SingleOutputStreamOperator<Tuple2<String,String>> price = objectSingleOutputStreamOperator.keyBy(value -> value.f0).reduce(new ReduceFunction<Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> reduce(Tuple2<String, String> value1, Tuple2<String, String> value2) throws Exception {
                if(value1.f1.compareTo(value2.f1)>0){
                    return value1;
                }
                return value2;
            }
        });
     //   price.print();

        environment.execute();

    }
}
